package com.ssa;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;

/**
 * ytx Consumer
 *
 * @author <a href="mailto:chenlsh@ytx.com">陈录顺</a>
 * @since 1.0.0
 */
public class Consumer {

    static String topic = "test1";

    static volatile boolean running = true;

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.10.151:9092");
        properties.setProperty("group.id", "test");
        properties.setProperty("enable.auto.commit", "true");
        properties.setProperty("auto.commit.interval.ms", "1000");
        properties.setProperty("session.timeout.ms", "30000");
        properties.setProperty("heartbeat.interval.ms", "2000");
        //  properties.setProperty("group.min.session.timeout.ms", "10000");
        //  properties.setProperty("group.max.session.timeout.ms", "1000");


        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");


        KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singleton(topic));

       // consumer.commitAsync();
        consumer.commitAsync(new OffsetCommitCallback() {
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                //
                System.out.println(offsets+"\t");
            }
        });

        while (running) {


            ConsumerRecords<Integer, String> record = consumer.poll(1000);

            record.forEach(it -> {
                System.out.println(it.key() + "\t" + it.offset() + "\t" + it.value());
            });
        }
    }
}
